package ex.datastream.functions.function;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * 按key,value形式对key数据的value进行累加
 */
public class ReduceFunc02 implements ReduceFunction<Tuple2<String, Long>> {

    @Override
    public Tuple2<String, Long> reduce(Tuple2<String, Long> v1, Tuple2<String, Long> v2) {
        return new Tuple2<>(v1.f0, v1.f1 + v2.f1);
    }

    /**
     * 输入
     * input:1> (a,1)
     * input:1> (a,3)
     * input:1> (a,23)
     * input:2> (b,2)
     * input:2> (c,13)
     * input:2> (d,33)
     * 输出
     * 2> (a,1)
     * 2> (a,4)
     * 2> (a,27)
     * 2> (d,33)
     * 1> (b,2)
     * 1> (c,13)
     */
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Tuple2<String, Long>> input = env.fromElements(
                Tuple2.of("a", 1L), Tuple2.of("b", 2L), Tuple2.of("a", 3L), Tuple2.of("c", 13L), Tuple2.of("a", 23L), Tuple2.of("d", 33L));
        input.printToErr("input");

        input.keyBy(value -> value.f0).reduce(new ReduceFunc02()).print();
        env.execute();
    }

}
